package com.googlecode.quillen.util;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.ArrayList;

/**
 * Created by IntelliJ IDEA.
 * User: greg
 * Date: Dec 2, 2008
 * Time: 10:43:56 PM
 */
public class WorkQueue
{
    private static final int DEFAULT_CAPACITY = 50;
    private static final int DEFAULT_NUM_WORKERS = 30;

    private final BlockingQueue<WorkItem> queue;
    private final Worker[] workers;
    private final ExceptionHandler exceptionHandler;
    private boolean frozen = false;
    private boolean aborted = false;

    public WorkQueue(int numWorkers, int capacity, ExceptionHandler exceptionHandler)
    {
        if (numWorkers <= 0)
        {
            throw new IllegalArgumentException("numWorkers must be greater than zero");
        }

        if (capacity <= 0)
        {
            throw new IllegalArgumentException("capacity must be greater than zero");
        }

        queue = new ArrayBlockingQueue<WorkItem>(capacity);
        workers = new Worker[numWorkers];

        for (int i = 0; i < numWorkers; i++)
        {
            this.workers[i] = new Worker();
            this.workers[i].start();
        }

        this.exceptionHandler = exceptionHandler;
    }

    public WorkQueue(int numWorkers, ExceptionHandler exceptionHandler)
    {
        this(numWorkers, DEFAULT_CAPACITY, exceptionHandler);
    }

    public WorkQueue(ExceptionHandler exceptionHandler)
    {
        this(DEFAULT_NUM_WORKERS, DEFAULT_CAPACITY, exceptionHandler);
    }

    @SuppressWarnings("InfiniteLoopStatement")
    public void enqueue(WorkItem workItem) throws WorkQueueAbortedException
    {
        if (frozen)
        {
            throw new IllegalStateException("cannot enqueue work item after freezing work queue");
        }
        else if (aborted)
        {
            throw new WorkQueueAbortedException();
        }
        else
        {
            while (true)
            {
                try
                {
                    queue.put(workItem);
                    return;
                }
                catch (InterruptedException e)
                {
                    // Ignore and retry
                }
            }
        }
    }

    @SuppressWarnings("InfiniteLoopStatement")
    public void freeze()
    {
        while (true)
        {
            try
            {
                frozen = true;

                //noinspection ForLoopReplaceableByForEach
                for (int i = 0; i < workers.length; i++)
                {
                    queue.put(WorkItem.TERMINATION_WORK_ITEM);
                }

                return;
            }
            catch (InterruptedException e)
            {
                // Ignore and retry
            }
        }
    }

    @SuppressWarnings("InfiniteLoopStatement")
    public void blockUntilEmpty() throws WorkQueueAbortedException
    {
        if (frozen)
        {
            while (true)
            {
                try
                {
                    joinAllWorkers();

                    if (aborted)
                    {
                        throw new WorkQueueAbortedException();
                    }

                    if (!queue.isEmpty())
                    {
                        throw new IllegalStateException("the queue is not empty but all the workers are done and work queue is not aborted");
                    }

                    return;
                }
                catch (InterruptedException ex)
                {
                    // Ignore and retry
                }
            }
        }
        else
        {
            throw new IllegalStateException("must freeze work queue before calling blockUntilEmpty");
        }
    }

    public void freezeAndBlockUntilEmpty() throws WorkQueueAbortedException
    {
        freeze();
        blockUntilEmpty();
    }

    private void joinAllWorkers() throws InterruptedException
    {
        for (Worker worker : workers)
        {
            worker.join();
        }
    }

    private void interruptAllWorkersExcept(Thread except)
    {
        for (Worker worker : workers)
        {
            if (worker != except)
            {
                worker.interrupt();
            }
        }
    }

    public int size()
    {
        return queue.size();
    }

    public boolean isEmpty()
    {
        return size() <= 0;
    }

    public void abort()
    {
        aborted = true;
        interruptAllWorkersExcept(Thread.currentThread());
        queue.drainTo(new ArrayList<WorkItem>(queue.size()));
    }

    private class Worker extends Thread
    {
        public void run()
        {
            while (true)
            {
                try
                {
                    WorkItem workItem = queue.take();
                    if (workItem == WorkItem.TERMINATION_WORK_ITEM)
                    {
                        return;
                    }
                    else
                    {
                        workItem.run();
                    }
                }
                catch (InterruptedException ex)
                {
                    return;
                }
                catch (Exception ex)
                {
                    if (exceptionHandler.handleException(ex))
                    {
                        abort();
                        return;
                    }
                }
            }
        }
    }
}
